package david.java.flink.datasources;

import david.java.flink.pojo.User;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import scala.Tuple3;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @Description:
 * @Author: Zhaodawei
 * @Date: Create in 下午6:33 2020/12/14
 */
public class CustomUserSource implements SourceFunction<User> {
    private volatile boolean isRunning = true;
    private List<String> nameList = new ArrayList<>(Arrays.asList("汝庚老师", "王嫂", "范大侠"));
    Random random = new Random();

    private boolean printData;

    public CustomUserSource(boolean printData) {
        this.printData = printData;
    }

    public CustomUserSource() {
        this.printData = false;
    }

    @Override
    public void run(SourceContext<User> sourceContext) throws Exception {
        while (isRunning) {
            int index = random.nextInt(3);
            String name = nameList.get(index);
            int value = random.nextInt(10);
            long currentTimestamp = LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

            User user = new User();
            user.setName(name);
            user.setAge(value);
            user.setTimestamp(currentTimestamp);
            if(printData) {
                System.out.println("datasource ===> " + user);
            }
            Thread.sleep(1000);
            sourceContext.collect(user);
        }
    }

    @Override
    public void cancel() {
        isRunning = !isRunning;
    }

    public static void main(String[] args) {
    }
}
